/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

import org.apache.log4j.Logger;
import org.zeromq.ZContext;
import org.zeromq.ZLoop;
import org.zeromq.ZMQ; 
import org.zeromq.ZMsg;
import org.zeromq.ZLoop.IZLoopHandler;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

public class ZmqConsumerS extends ZmqConsumerBase implements Runnable {
    private Socket client = null;
	private int count = 0;
    private int retriesLeft;
    private byte[] message;
    private long offset;
    private ConsumerConnector consumer;
    
    private static Logger LOG = Logger.getLogger(ZmqConsumerS.class.getName());
    
    public void ack(int status) {
    	if(status == UNREACHABLE || status == TIMEOUT || status == INTERRUPTED)
			errHandler.OnError(status, offset);
    	
    	count++;
    	if (count >= bufferSize) {
    		consumer.commitOffsets();
    		count = 0;
    	}
    }

    public ZmqConsumerS(String topic, ConsumerConfig config, 
    	int bufferSize, ZContext ctx, String conn, 
    		int timeout, int retry, ErrorHandler errHandler) {
    	super(topic, config, bufferSize, ctx, conn, timeout, retry, errHandler);
    }
    
    private void connect() {
    	disconnect();
    	client = ctx.createSocket(ZMQ.DEALER);
    	id = CSBUtils.sockId();
    	client.setIdentity(id.getBytes());
    	client.connect(conn);
        LOG.info(id + ": connected to router via " + conn + "...");
    }
    
    private void disconnect() {
    	if (client != null)
    		ctx.destroySocket(client);
    }

    private IZLoopHandler resend = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
        	if (--retriesLeft <= 0) {
        		connect();
        		ack(TIMEOUT);
        		return -1;
        	}
        	
			LOG.error(id + ": no response from router, resending...");
			client.send(message);
            return 0;
        }
    };
    
    private IZLoopHandler proc = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {        	
			ZMsg msg = ZMsg.recvMsg(client);
			if (msg == null) {
				LOG.error(id + ": receiving interrupted, closing...");
        		connect();
        		ack(INTERRUPTED);
				return -1;    
			}
            
			msg.unwrap();
			String reply = new String(msg.pop().getData());
			if (reply.equals(CSB.WORKER_ACK)) { 
				msg.destroy();                
				ack(ACKNOWLEDGED); 
				return -1;
			} else if (reply.equals(CSB.ROUTER_UNREACHABLE)) {
				msg.destroy();
				ack(UNREACHABLE);
				return -1;
			} else {			
				LOG.warn(id + ": malformed reply from router: " + reply);
			}
			return 0;
        }
    };
    
    public void run() {
    	consumer = kafka.consumer.Consumer.
            	createJavaConsumerConnector(config);
        Map<String, Integer> topicCountMap = 
        		new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = 
        		consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream =  
        		consumerMap.get(topic).get(0);
        ConsumerIterator<byte[], byte[]> it = stream.iterator();
        
    	connect();
        LOG.info(id + ": consumer created, topic: " + topic);
        
        count = 0;
        while (!Thread.currentThread().isInterrupted()) {
            retriesLeft = retry;

        	if(it.hasNext()) {
        		MessageAndMetadata<byte[], byte[]> mam = it.next();
            	message = mam.message();
            	offset = mam.offset();
            	
            	String msg = new String(message);
            	if (msg.startsWith("%")) {
            		int index = msg.indexOf("%", 1);
            		String rid = msg.substring(1, index - 1);            		
            		msg = msg.substring(index);
                	client.sendMore(rid.getBytes());
                	client.send(msg.getBytes());
            	} else {
            		client.send(message);
            	}
            }        	
        	
        	ZLoop loop = new ZLoop();
    		PollItem poller = new PollItem(client, Poller.POLLIN);
    		loop.addPoller(poller, proc, null);
    		loop.addTimer(timeout, retry, resend, null);        		
    		loop.start();
    		loop.destroy();
        }
        
        disconnect();
        consumer.shutdown();
    }    
}

